package com.pnnd.stream;

import java.io.IOException;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class NdArrayInputStream extends MultiplySegmentInputStream {


    public NdArrayInputStream(List<MemorySegment> segments) {
        super(segments);
    }

    private Queue<Operator> operators;
    private Thread operatorThread;

    public void addOperator(Operator o){
        this.operators.add(o);

        this.invokeOperator();
    }

    public void removeOperator(Operator o){
        if(this.operators.contains(o)){
            this.operators.remove(o);
        }else{
            throw new RuntimeException("operator has running");
        }
    }

    private void invokeOperator(){
        if(this.operatorThread == null){
            this.operatorThread = new Thread(() -> {

                Operator operator = NdArrayInputStream.this.operators.poll();

                while(operator!=null){

                    if(operator instanceof ReadOperator){
                        ReadOperator readOperator = (ReadOperator)operator;
                        try {
                            int len = NdArrayInputStream.super.read(readOperator.getTarget(),readOperator.getOffset(),readOperator.getLen());

                            OperatorResult result = new OperatorResult(true);

                            result.operator = readOperator;
                            result.setReadLen(len);

                            readOperator.setResult(result);
                        }catch (IOException e){
                            readOperator.setResult(new OperatorResult(e,false));
                        }
                    }

                    if(operator instanceof SeekOperator){
                        SeekOperator seekOperator = (SeekOperator)operator;
                        try {
                            NdArrayInputStream.super.seek(seekOperator.getSeek());

                            OperatorResult result = new OperatorResult(true);

                            result.operator = seekOperator;

                            seekOperator.setResult(result);
                        }catch (Exception e){
                            seekOperator.setResult(new OperatorResult(e,false));
                        }
                    }

                    if(operator instanceof MarkOperator){
                        MarkOperator markOperator = (MarkOperator)operator;
                        try {
                            NdArrayInputStream.super.mark(markOperator.getMark());

                            OperatorResult result = new OperatorResult(true);

                            result.operator = markOperator;

                            markOperator.setResult(result);
                        }catch (Exception e){
                            markOperator.setResult(new OperatorResult(e,false));
                        }
                    }

                    if(operator instanceof ResetOperator){
                        ResetOperator resetOperator = (ResetOperator)operator;
                        try {
                            NdArrayInputStream.super.reset();

                            OperatorResult result = new OperatorResult(true);

                            result.operator = resetOperator;

                            resetOperator.setResult(result);
                        }catch (Exception e){
                            resetOperator.setResult(new OperatorResult(e,false));
                        }
                    }

                    operator = NdArrayInputStream.this.operators.poll();
                }


                NdArrayInputStream.this.operatorThread = null;
            });
        }
    }

    public static abstract class Operator{

        private String id;
        private NdArrayInputStream ndArrayInputStream;

        public Operator(){
            this.id = UUID.randomUUID().toString();
        }

        @Override
        public int hashCode(){
            return this.id.hashCode();
        }

        @Override
        public boolean equals(Object other){
            if(other instanceof Operator){
                return ((Operator) other).id.equals(this.id);
            }

            return false;
        }

        public abstract OperatorResult getResult();

        public void cancel(){
            this.ndArrayInputStream.removeOperator(this);
        }
    }

    public static class OperatorResult{
        private int readLen;
        private boolean success;
        private Exception exception;

        private Operator operator;


        public OperatorResult(Exception e,boolean success){
            this.exception = e;
            this.success = success;
        }

        public OperatorResult(boolean success){
            this.success = success;
        }

        public Exception getException() {
            return exception;
        }

        public void setException(Exception exception) {
            this.exception = exception;
        }

        public boolean isSuccess() {
            return success;
        }

        public void setSuccess(boolean success) {
            this.success = success;
        }

        public int getReadLen() {
            return readLen;
        }

        public void setReadLen(int readLen) {
            this.readLen = readLen;
        }

        public byte[] getTarget() {

            if(this.operator instanceof ReadOperator){
                return ((ReadOperator) this.operator).getTarget();
            }

            return null;
        }

        public int getOffset() {
            if(this.operator instanceof ReadOperator){
                return ((ReadOperator) this.operator).getOffset();
            }

            return 0;
        }

        public int getLen() {
            if(this.operator instanceof ReadOperator){
                return ((ReadOperator) this.operator).getLen();
            }

            return 0;
        }

        public Operator getOperator() {
            return operator;
        }

        public void setOperator(Operator operator) {
            this.operator = operator;
        }
    }

    public static class ReadOperator extends Operator{

        private Object source;
        private InputCallback callback;
        private OperatorResult result;
        private byte[] target;
        private int offset;
        private int len;

        @Override
        public OperatorResult getResult() {
            return this.result;
        }

        public void setResult(OperatorResult result){
            this.result = result;
        }

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public InputCallback getCallback() {
            return callback;
        }

        public void setCallback(InputCallback callback) {
            this.callback = callback;
        }

        public byte[] getTarget() {
            return target;
        }

        public void setTarget(byte[] target) {
            this.target = target;
        }

        public int getOffset() {
            return offset;
        }

        public void setOffset(int offset) {
            this.offset = offset;
        }

        public int getLen() {
            return len;
        }

        public void setLen(int len) {
            this.len = len;
        }
    }

    public static class SeekOperator extends Operator{
        private Object source;
        private InputCallback callback;
        private OperatorResult result;
        private int seek;

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public InputCallback getCallback() {
            return callback;
        }

        public void setCallback(InputCallback callback) {
            this.callback = callback;
        }

        @Override
        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }

        public int getSeek() {
            return seek;
        }

        public void setSeek(int seek) {
            this.seek = seek;
        }
    }

    public static class MarkOperator extends Operator{
        private Object source;
        private InputCallback callback;
        private OperatorResult result;
        private int mark;

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public InputCallback getCallback() {
            return callback;
        }

        public void setCallback(InputCallback callback) {
            this.callback = callback;
        }

        @Override
        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }


        public int getMark() {
            return mark;
        }

        public void setMark(int mark) {
            this.mark = mark;
        }
    }

    public static class ResetOperator extends Operator {
        private Object source;
        private InputCallback callback;
        private OperatorResult result;

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public InputCallback getCallback() {
            return callback;
        }

        public void setCallback(InputCallback callback) {
            this.callback = callback;
        }

        @Override
        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }
    }

    class OperatorFuture implements Future<OperatorResult> {

        private Operator operator;
        private boolean canceled;

        public OperatorFuture(Operator operator){
            this.operator = operator;
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            try {
                this.operator.cancel();
                this.canceled = true;
                return true;
            }catch (RuntimeException e){
                return false;
            }
        }

        @Override
        public boolean isCancelled() {
            return this.canceled;
        }

        @Override
        public boolean isDone() {
            return this.operator.getResult() != null;
        }

        @Override
        public OperatorResult get() throws InterruptedException, ExecutionException {

            return this.operator.getResult();
        }

        @Override
        public OperatorResult get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {

            this.await(timeout,unit);

            return this.get();
        }

        public void await(long timeout, TimeUnit unit){
            if(timeout == -1){
                while (isDone());

                return;
            }

            long currentTime = System.currentTimeMillis();

            while((System.currentTimeMillis()-currentTime)>0 || !isDone());

            return;
        }
    }

    public Future<OperatorResult> asyncRead(Object source,InputCallback callback){
        ReadOperator readOperator = new ReadOperator();

        readOperator.setLen(1);
        readOperator.setOffset(0);
        readOperator.setTarget(new byte[1]);

        readOperator.setSource(source);
        readOperator.setCallback(callback);

        OperatorFuture future = new OperatorFuture(readOperator);

        this.addOperator(readOperator);

        return future;
    }

    public Future<OperatorResult> asyncRead(){
        return this.asyncRead(null,null);
    }

    public Future<OperatorResult> asyncRead(byte[] target){
        return this.asyncRead(target,0,target.length);
    }

    public Future<OperatorResult> asyncRead(byte[] target,Object source,InputCallback cb){
        return this.asyncRead(target,0,target.length,source,cb);
    }

    public Future<OperatorResult> asyncRead(byte[] target,int off,int len){
        return this.asyncRead(target,off,len,null,null);
    }

    public Future<OperatorResult> asyncRead(byte[] target,int off,int len,Object source,InputCallback callback){
        ReadOperator readOperator = new ReadOperator();

        readOperator.setLen(len);
        readOperator.setOffset(off);
        readOperator.setTarget(target);

        readOperator.setSource(source);
        readOperator.setCallback(callback);

        OperatorFuture future = new OperatorFuture(readOperator);

        this.addOperator(readOperator);

        return future;
    }

    public Future<OperatorResult> asyncMark(int value){
        return this.asyncMark(value,null,null);
    }

    public Future<OperatorResult> asyncMark(int value,Object source,InputCallback cb){
        MarkOperator markOperator = new MarkOperator();

        markOperator.setSource(source);
        markOperator.setCallback(cb);
        markOperator.setMark(value);

        OperatorFuture future = new OperatorFuture(markOperator);

        this.addOperator(markOperator);

        return future;
    }

    public Future<OperatorResult> asyncReset(){
        return this.asyncReset(null,null);
    }

    public Future<OperatorResult> asyncReset(Object source,InputCallback cb){
        ResetOperator resetOperator = new ResetOperator();

        resetOperator.setSource(source);
        resetOperator.setCallback(cb);

        OperatorFuture future = new OperatorFuture(resetOperator);

        this.addOperator(resetOperator);

        return future;
    }

    public Future<OperatorResult> asyncSeek(int value){
        return this.asyncSeek(value,null,null);
    }

    public Future<OperatorResult> asyncSeek(int value,Object source,InputCallback cb){
        SeekOperator seekOperator = new SeekOperator();

        seekOperator.setSource(source);
        seekOperator.setCallback(cb);
        seekOperator.setSeek(value);

        OperatorFuture future = new OperatorFuture(seekOperator);

        this.addOperator(seekOperator);

        return future;
    }

    @Override
    public int read() throws IOException{
        Future<OperatorResult> future = this.asyncRead();

        try {
            future.get(-1,TimeUnit.MILLISECONDS);

            return future.get().getTarget()[0];
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    @Override
    public int read(byte[] bs) throws IOException{
        Future<OperatorResult> future = this.asyncRead(bs);

        try {
            future.get(-1,TimeUnit.MILLISECONDS);

            return future.get().getReadLen();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    @Override
    public int read(byte[] bs,int off,int len) throws IOException{
        Future<OperatorResult> future = this.asyncRead(bs,off,len);

        try {
            future.get(-1,TimeUnit.MILLISECONDS);

            return future.get().getReadLen();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    @Override
    public void mark(int value) {
        Future<OperatorResult> future = this.asyncMark(value);

        try {
            future.get(-1,TimeUnit.MILLISECONDS);;
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    @Override
    public void seek(int value) {
        Future<OperatorResult> future = this.asyncSeek(value);

        try {
            future.get(-1,TimeUnit.MILLISECONDS);;
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    @Override
    public void reset() {
        Future<OperatorResult> future = this.asyncReset();

        try {
            future.get(-1,TimeUnit.MILLISECONDS);;
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }
}
